import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
import static org.apache.flink.configuration.TaskManagerOptions.MEMORY_SEGMENT_SIZE;
import static org.apache.flink.configuration.TaskManagerOptions.NETWORK_MEMORY_MAX;

public class FlinkCDC {
//    public static void main(String[] args) throws Exception {
//        //1.获取Flink 执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        env.setParallelism(1);
//        Configuration configuration = new Configuration();
//        configuration.set(NETWORK_MEMORY_MAX, MemorySize.parse("1G"));
//        configuration.set(MEMORY_SEGMENT_SIZE, MemorySize.parse("8kb"));
//        configuration.set(DEFAULT_PARALLELISM, 1);
//
//        //2.通过FlinkCDC构建SourceFunction
//        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
//                .hostname("127.0.0.1")
//                .port(3306)
//                .username("root")
//                .password("123456")
//                .databaseList("cims")
//                .tableList("cims.student")
////                .startupOptions(StartupOptions.initial())
//                .startupOptions(StartupOptions.latest())
//                .deserializer(new StringDebeziumDeserializationSchema())
//                .build();
//
//        DataStreamSource<String> dataStreamSource = env.addSource(mysqlSource);
//
//        //3.数据打印
//        dataStreamSource.process(new ProcessFunction<String, Object>() {
//            @Override
//            public void processElement(String s, ProcessFunction<String, Object>.Context context, Collector<Object> collector) throws Exception {
//                System.out.println(s);
//            }
//        });
//
//        //4.启动任务
//        env.execute("FlinkCDC");
//    }
}
